package com.jiunuan.live.user.provider.config;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

@Configuration
@Slf4j
public class RocketMQProducerConfig {
    @Resource
    private RocketMQProducerProperties rocketMQProducerProperties;
    @Value("${spring.application.name")
    private String applicationName;
    /**
     * 使MQProducer以Bean注入springboot容器
     * @return
     */
    @Bean
    public MQProducer mqProducer(){

        // 创建一个线程池，用于执行大量的异步任务
        ThreadPoolExecutor asyncThreadPoolExecutor = new ThreadPoolExecutor(
                // 核心线程数，即始终维持的最小线程数
                100,
                // 最大线程数，超过这个数量的任务将被排队等待
                150,
                // 线程空闲的最长时间，超过这个时间的空闲线程将被终止
                3,
                // 时间单位
                TimeUnit.MINUTES,
                // 任务队列，用于存储等待执行的任务
                new ArrayBlockingQueue<>(1000),
                // 线程工厂，用于创建新的线程
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        // 创建一个新的线程并设置其名称
                        Thread thread = new Thread(r);
                        // 设置线程名称，方便问题定位和性能监控
                        thread.setName(applicationName+":rmq-producer"+
                                ThreadLocalRandom.current().nextInt(1000));
                        return thread;
                    }
                }
        );


        // 创建一个默认的RocketMQ生产者实例
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        try {
            // 设置名称服务器地址，这是连接RocketMQ服务的关键配置
            defaultMQProducer.setNamesrvAddr(rocketMQProducerProperties.getNameSrv());
            // 设置生产者组名，用于标识同一组内的生产者
            defaultMQProducer.setProducerGroup(rocketMQProducerProperties.getGroupName());
            // 设置当消息发送失败时的重试次数
            defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProducerProperties.getRetryTimes());
            // 设置当异步发送消息失败时的重试次数
            defaultMQProducer.setRetryTimesWhenSendAsyncFailed(rocketMQProducerProperties.getRetryTimes());
            // 开启当消息未能正确存储时重试发送到另一个Broker的选项
            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
            // 设置异步发送消息所使用的线程池执行器
            defaultMQProducer.setAsyncSenderExecutor(asyncThreadPoolExecutor);
            // 启动生产者实例，使其准备就绪可以发送消息
            defaultMQProducer.start();
            log.info("RocketMQProducer start success,namesrvAddr is {}", rocketMQProducerProperties.getNameSrv());

        } catch (MQClientException e) {
            // 如果在启动生产者过程中遇到异常，则抛出运行时异常
            throw new RuntimeException(e);
        }
        // 返回启动后的生产者实例
        return defaultMQProducer;

    }
}
